using Aliphant.Common;

using GRPCDemo.Protocol;

namespace Common.NetWork
{
    public class MessageDistributer : MessageDistributer<object>
    {

    }

    public class MessageDistributer<T> : Singleton<MessageDistributer<T>>
    {
        class MessageArgs
        {
            public T? sender;
            public NetMessage? message;
        }

        private bool Runing = false;

        private int ActiveThreadCount = 0;
        private int ThreadCount = 0;
        private object locker = new object();

        private Queue<MessageArgs> messageQueue = new Queue<MessageArgs>();
        public delegate void MessageHandler<Tm>(T sender, Tm message);
        private Dictionary<string, Delegate?> messageHandlers = new Dictionary<string, Delegate?>();

        private AutoResetEvent threadEvent = new AutoResetEvent(true);


        public void Start(int ThreadNum)
        {
            this.ThreadCount = ThreadNum;
            if (this.ThreadCount < 1) this.ThreadCount = 1;
            if (this.ThreadCount > 1000) this.ThreadCount = 1000;
            Runing = true;

            for (int i = 0; i < this.ThreadCount; i++)
            {
                ThreadPool.QueueUserWorkItem(new WaitCallback(MessageDistribute));
            }
            while (ActiveThreadCount < this.ThreadCount)
            {
                Thread.Sleep(100);
            }
        }

        public void Stop()
        {
            Runing = false;
            while (ActiveThreadCount > 0)
            {
                threadEvent.Set();
            }
            Thread.Sleep(100);
        }

        public void Subscribe<Tm>(MessageHandler<Tm> handler)
        {
            string type = typeof(Tm).Name;
            if (!messageHandlers.ContainsKey(type))
            {
                messageHandlers[type] = null;
            }

            var tHnder = messageHandlers[type] as MessageHandler<Tm>;
            if (tHnder != null)
            {
                messageHandlers[type] = tHnder + handler;
            }
        }

        public void Unsubscribe<Tm>(MessageHandler<Tm> handler)
        {
            string type = typeof(Tm).Name;
            if (!messageHandlers.ContainsKey(type))
            {
                messageHandlers[type] = null;
            }

            var tHnder = messageHandlers[type] as MessageHandler<Tm>;
            if (tHnder != null)
            {
                messageHandlers[type] = tHnder - handler;
            }
        }

        public void ReceiveMessage(T sender, NetMessage message)
        {
            lock (locker)
            {
                this.messageQueue.Enqueue(new MessageArgs() { sender = sender, message = message });
            }
            threadEvent.Set();
        }

        public void RaiseEvent<Tm>(T sender, Tm msg)
        {
            string type = msg?.GetType().Name ?? string.Empty;
            if (messageHandlers.ContainsKey(type))
            {
                var handler = messageHandlers[type] as MessageHandler<Tm>;
                if (handler != null)
                {
                    try
                    {
                        handler(sender, msg);
                    }
                    catch (Exception ex)
                    {
                        Logger.ErrorFormat("Message handler exception:{0}", ex);
                        //if (ThrowException)
                        //    throw ex;
                    }
                }
            }
        }

        /// <summary>
        /// 一次性分发队列中的所有消息
        /// </summary>
        public void Distribute()
        {
            if (this.messageQueue.Count == 0)
            {
                return;
            }

            while (this.messageQueue.Count > 0)
            {
                MessageArgs package = this.messageQueue.Dequeue();
                if (package == null ||
                    package.message == null ||
                    package.sender == null)
                    continue;

                if (package.message.Request != null)
                    MessageDispatch<T>.Instance.Dispatch(package.sender, package.message.Request);
                if (package.message.Response != null)
                    MessageDispatch<T>.Instance.Dispatch(package.sender, package.message.Response);
            }
        }

        private void MessageDistribute(object? state)
        {
            Logger.Info("MessageDistribute thread start");
            try
            {
                ActiveThreadCount = Interlocked.Increment(ref ActiveThreadCount);
                while (Runing)
                {
                    if (this.messageQueue.Count == 0)
                    {
                        threadEvent.WaitOne();
                        continue;
                    }
                    MessageArgs package;
                    lock (locker)
                    {
                        package = messageQueue.Dequeue();
                    }
                    // 消息分发

                    if (package == null ||
                        package.message == null ||
                        package.sender == null)
                        continue;

                    if (package.message.Request != null)
                        MessageDispatch<T>.Instance.Dispatch(package.sender, package.message.Request);
                    if (package.message.Response != null)
                        MessageDispatch<T>.Instance.Dispatch(package.sender, package.message.Response);
                }
            }
            catch
            {

            }
            finally
            {
                ActiveThreadCount = Interlocked.Decrement(ref ActiveThreadCount);
            }
        }
    }
}
